29. 农产品交易销售趋势与产品结构分析

import pandas as pd  # 导入表格库
def describe_platform_agriculture(agri_transactions):  # 定义不声称预测验证的描述接口
    required={'order_id','order_date','category','quantity','sales_amount'}  # 对齐平台实际字段
    if not required<=set(agri_transactions.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时暂停分析
    transactions=agri_transactions.dropna(subset=list(required)).drop_duplicates().copy()  # 清除关键缺失与完全重复行
    if transactions.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空样本时停止
    transactions['order_date']=pd.to_datetime(transactions['order_date'],errors='coerce')  # 统一交易日期
    if transactions['order_date'].isna().any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无效日期停止
    monthly=transactions.set_index('order_date').resample('ME').agg(orders=('order_id','nunique'),sales=('sales_amount','sum'))  # 生成完整月度序列
    category=transactions.groupby('category').agg(orders=('order_id','nunique'),quantity=('quantity','sum'),sales=('sales_amount','sum'))  # 生成品类结构与分母
    complete=not monthly[['orders','sales']].eq(0).any().any()  # 检查连续月份是否均有非零观测
    status={'mode':'DESCRIPTIVE_ONLY','months':len(monthly),'complete_nonzero_months':complete,'forecast_ready':len(monthly)>=18 and complete}  # 明示回退边界
    if abs(category['sales'].sum()-transactions['sales_amount'].sum())>=1e-8: raise AssertionError('sales reconciliation failed')  # 核对品类销售额守恒
    return transactions,monthly,category,status  # 返回描述结果而不声称预测验证

from sklearn.linear_model import LinearRegression  # 导入趋势回归模型
def forecast_platform_agriculture(agri_transactions):  # 定义平台交易完整预测接口
    transactions,monthly,category,status=describe_platform_agriculture(agri_transactions)  # 先生成诚实描述回退
    if not status['forecast_ready']: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 预测期不足时停止
    monthly['seasonal_baseline']=monthly['sales'].shift(12)  # 构造上年同月基线
    monthly['trend_index']=range(len(monthly))  # 构造时间趋势特征
    model_frame=monthly.dropna(subset=['seasonal_baseline']).copy()  # 固定可比较样本
    validation_size=min(12,max(2,len(model_frame)//3))  # 事先设定末段验证长度
    if len(model_frame)-validation_size<4: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 明确训练窗下限
    training=model_frame.iloc[:-validation_size]  # 按时间切分训练窗
    validation=model_frame.iloc[-validation_size:].copy()  # 隔离同一验证窗
    trend_model=LinearRegression().fit(training[['trend_index']],training['sales'])  # 仅用训练窗拟合趋势
    validation['trend_forecast']=trend_model.predict(validation[['trend_index']])  # 生成验证期预测
    validation['baseline_ape']=(validation['sales']-validation['seasonal_baseline']).abs()/validation['sales'].abs()  # 计算基线逐期误差
    validation['trend_ape']=(validation['sales']-validation['trend_forecast']).abs()/validation['sales'].abs()  # 计算趋势逐期误差
    decision={'threshold':0.15,'threshold_status':'课堂误差情景，待业务规定批准','owner':'农产品运营负责人','action':'仅在趋势MAPE优于基线且过最低要求时小范围试用','recheck_when':'字段发生变化、缺月或MAPE超阈值','recheck_date':'下月第5个工作日'}  # 建立完整决策说明信息
    decision['status']='仅作课堂比较' if validation['trend_ape'].mean()<validation['baseline_ape'].mean() and validation['trend_ape'].mean()<=decision['threshold'] else '暂不采用'  # 同一时间段检查要求
    return monthly,category,validation,decision  # 返回趋势、结构、误差与检查要求

def select_latest_disclosure(frame,value_fields):  # 统一财报披露版本选择约定
    required={'ts_code','end_date',*value_fields}; versions=[c for c in ['ann_date','f_ann_date'] if c in frame.columns]  # 固定键与可接受版本字段
    if not required<=set(frame.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
    work=frame.dropna(subset=list(required)).copy(); work['end_date']=pd.to_datetime(work['end_date'],errors='coerce')  # 清理必需字段
    for column in versions: work[column]=pd.to_datetime(work[column],errors='coerce')  # 解析披露日期
    keys=['ts_code','end_date']; work=work.dropna(subset=['end_date']); duplicate_mask=work.duplicated(keys,keep=False); before=int(duplicate_mask.sum())  # 统计选择前重复
    if before and (not versions or work.loc[duplicate_mask,versions].isna().all(axis=1).any()): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无版本依据时停止
    work['_version']=work[versions].max(axis=1) if versions else pd.NaT  # 合并双版本字段为排序键
    latest_version=work.groupby(keys)['_version'].transform('max') if versions else pd.Series(pd.NaT,index=work.index)  # 为每个公司报告期定位最晚披露时点
    latest_rows=work.loc[~duplicate_mask | work['_version'].eq(latest_version)].copy()  # 仅保留单例或并列最晚版本
    tied_mask=latest_rows.duplicated(keys,keep=False); tied_latest_rows=int(tied_mask.sum())  # 统计并列最晚版本行
    conflicting_latest_keys=int(latest_rows.loc[tied_mask].groupby(keys,dropna=False)[value_fields].nunique(dropna=False).gt(1).any(axis=1).sum()) if tied_latest_rows else 0  # 统计并列且数值冲突的披露键
    if conflicting_latest_keys: raise ValueError(f'输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 并列最新值冲突时请先检查数据后再继续
    selected=latest_rows.drop_duplicates(keys+value_fields).drop_duplicates(keys).drop(columns='_version',errors='ignore')  # 仅合并数值完全一致的并列行
    after=int(selected.duplicated(keys).sum()); audit={'duplicate_rows_before':before,'tied_latest_rows':tied_latest_rows,'conflicting_latest_keys':conflicting_latest_keys,'duplicate_rows_after':after,'version_fields':versions}  # 返回并列与冲突检查
    if after: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 选择后仍重复则停止
    return selected,audit  # 返回唯一版本与检查依据

from pathlib import Path  # 导入路径工具绑定规定资产
import pandas as pd  # 导入表格库以标准化HDF字段
import numpy as np  # 导入有限值与覆盖率检查工具
from sklearn.linear_model import LinearRegression  # 导入趋势回归
root=Path('/home/ubuntu/r2_data_mount/data/stock')  # 指定股票数据目录
basic_path,financial_path=root/'stock_basic_data.h5',root/'financial_statement.h5'  # 绑定可用HDF公司与财报快照
if not basic_path.exists() or not financial_path.exists(): raise FileNotFoundError('未找到课程股票数据文件，请从课程数据下载入口获取')  # 提示读者下载文件
basic=pd.read_hdf(basic_path,key='stock_basic_info',columns=['order_book_id','symbol','industry_name','province']).rename(columns={'order_book_id':'raw_code','symbol':'name','industry_name':'industry','province':'area'})  # 只读取公司筛选字段并标准化
basic['ts_code']=basic['raw_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一证券代码后缀
basic['area']=basic['area'].astype('string').str.replace(r'[省市]$','',regex=True)  # 统一省市简称
required_basic={'ts_code','name','industry','area'}  # 定义公司字段结构
required_financial={'ts_code','end_date','revenue','operate_profit'}  # 定义财报字段结构
focus=basic.loc[basic['industry'].str.contains('农|林|牧|渔',na=False) & basic['area'].isin(['上海','江苏','浙江','安徽'])]  # 筛选长三角农业企业
if focus.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空对象时停止
focus_raw_codes=focus['raw_code'].tolist()  # 保留HDF原始代码用于高效行筛选
financial=pd.read_hdf(financial_path,key='financial_data',where='order_book_id in focus_raw_codes',columns=['order_book_id','quarter','info_date','operating_revenue','profit_from_operation']).rename(columns={'order_book_id':'ts_code','quarter':'end_date','info_date':'ann_date','operating_revenue':'revenue','profit_from_operation':'operate_profit'})  # 按对象读取并标准化财报字段
financial['ts_code']=financial['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一财报证券代码后缀
financial['end_date']=pd.PeriodIndex(financial['end_date'].str.upper(),freq='Q').end_time.normalize()  # 将季度键转换为报告期末日期
if not required_basic<=set(basic.columns) or not required_financial<=set(financial.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 标准化后仍缺列时停止
financial,version_audit=select_latest_disclosure(financial,['revenue','operate_profit'])  # 统一保留最晚披露版本

panel=financial.merge(focus[['ts_code','name','area']],on='ts_code',how='inner',validate='many_to_one').copy()  # 合并对象标签
if focus.empty or panel.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空对象或空连接时停止
panel=panel.sort_values(['ts_code','end_date'])  # 固定公司时间顺序
panel['fiscal_year']=panel['end_date'].dt.year  # 构造财政年键
panel['fiscal_quarter']=panel['end_date'].dt.quarter  # 构造财政季度键
if panel.duplicated(['ts_code','fiscal_year','fiscal_quarter']).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 公司期间键不唯一时停止
REVENUE_CADENCE='YTD_CUMULATIVE'  # 绑定 financial_statement 的利润表年初至今累计口径
if REVENUE_CADENCE!='YTD_CUMULATIVE': raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 先确认收入统计方式再继续
prior_cumulative=panel[['ts_code','fiscal_year','fiscal_quarter','revenue','operate_profit']].copy()  # 建立上一季度累计值查找表
prior_cumulative['fiscal_quarter']=prior_cumulative['fiscal_quarter']+1  # 只把同年上一季度键前移一季
prior_cumulative=prior_cumulative.rename(columns={'revenue':'prior_cumulative_revenue','operate_profit':'prior_cumulative_operate_profit'})  # 标明累计前值
panel=panel.merge(prior_cumulative,on=['ts_code','fiscal_year','fiscal_quarter'],how='left',validate='one_to_one')  # 按公司、财年、季度精确连接
needs_prior=panel['fiscal_quarter'].gt(1)  # 二至四季度累计值必须有同年上一季度
missing_prior=needs_prior & panel[['prior_cumulative_revenue','prior_cumulative_operate_profit']].isna().any(axis=1)  # 统计无法解累计值
panel['quarter_revenue']=np.where(panel['fiscal_quarter'].eq(1),panel['revenue'],panel['revenue']-panel['prior_cumulative_revenue'])  # 精确还原单季收入
panel['quarter_operate_profit']=np.where(panel['fiscal_quarter'].eq(1),panel['operate_profit'],panel['operate_profit']-panel['prior_cumulative_operate_profit'])  # 精确还原单季营业利润
panel['quarter_key']=panel['end_date'].dt.to_period('Q-DEC')  # 从报告期日期构造连续日历季度键
quarter_counts=panel.loc[~missing_prior].groupby('quarter_key')['ts_code'].nunique().sort_index()  # 统计可观察公司覆盖
eligible_quarters=quarter_counts.loc[quarter_counts.ge(2)]  # 至少两家公司才允许定义行业季度
if eligible_quarters.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无可分析季度时停止

coverage_floor=0.80  # 事先设定终点季度覆盖率下限
candidate_companies=panel.loc[~missing_prior,'ts_code'].nunique()  # 固定覆盖率分母
eligible_end=quarter_counts.loc[quarter_counts.div(candidate_companies).ge(coverage_floor)]  # 找到达到覆盖率的季度
if eligible_end.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 终点不稳定时停止
window_end=eligible_end.index.max(); window_quarters=pd.period_range(end=window_end,periods=16,freq='Q-DEC')  # 固定连续十六季度观察窗以支持四季样本外验证
presence=panel.loc[~missing_prior & panel['quarter_key'].isin(window_quarters)].groupby('ts_code')['quarter_key'].nunique()  # 统计公司在固定窗内的完整性
stable_codes=presence.loc[presence.eq(len(window_quarters))].index  # 稳定样本只保留十六季均可观察公司
if len(stable_codes)<2: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 样本组成不足时停止
panel=panel.loc[panel['ts_code'].isin(stable_codes) & panel['quarter_key'].isin(window_quarters)].copy()  # 固定公司组成与期间窗口
composition=panel.groupby('quarter_key')['ts_code'].nunique().reindex(window_quarters,fill_value=0)  # 检查每季公司组成
if not composition.eq(len(stable_codes)).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 组成变化请先检查数据后再继续
if panel[['quarter_revenue','quarter_operate_profit']].isna().any().any() or not np.isfinite(panel[['quarter_revenue','quarter_operate_profit']].to_numpy()).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 转换值检查要求
if panel['quarter_revenue'].eq(0).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 单季分母检查要求
company_prior=panel[['ts_code','fiscal_year','fiscal_quarter','quarter_revenue']].rename(columns={'quarter_revenue':'seasonal_baseline'}).copy()  # 建立公司上年同季单季基线
company_prior['fiscal_year']=company_prior['fiscal_year']+1  # 将历史年键前移一年以匹配当期
panel=panel.merge(company_prior,on=['ts_code','fiscal_year','fiscal_quarter'],how='left',validate='one_to_one')  # 精确连接同公司上年同季基线
panel['baseline_ape']=(panel['quarter_revenue']-panel['seasonal_baseline']).abs()/panel['quarter_revenue'].abs()  # 计算同一时间段的基准方法误差
panel['operating_margin']=panel['quarter_operate_profit']/panel['quarter_revenue']  # 计算单季营业利润率检查要求指标
cohort_audit={'revenue_cadence':REVENUE_CADENCE,'candidate_companies':int(candidate_companies),'coverage_floor':coverage_floor,'window_start':str(window_quarters.min()),'window_end':str(window_quarters.max()),'stable_companies':int(len(stable_codes)),'excluded_companies':int(candidate_companies-len(stable_codes)),'companies_each_quarter':composition.astype(int).to_dict()}  # 发布口径、覆盖与固定组成检查

quarterly=panel.groupby(['fiscal_year','fiscal_quarter']).agg(end_date=('end_date','max'),revenue=('quarter_revenue','sum'),companies=('ts_code','nunique')).reset_index()  # 按精确日历季度汇总固定样本的单季收入
if quarterly.empty or quarterly['revenue'].eq(0).any() or not np.isfinite(quarterly['revenue']).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 汇总分母检查要求
industry_prior=quarterly[['fiscal_year','fiscal_quarter','revenue']].rename(columns={'revenue':'baseline'}).copy()  # 建立行业上年同季基线查找表
industry_prior['fiscal_year']=industry_prior['fiscal_year']+1  # 将历史年键前移一年以匹配当期
quarterly=quarterly.merge(industry_prior,on=['fiscal_year','fiscal_quarter'],how='left',validate='one_to_one').sort_values(['fiscal_year','fiscal_quarter'])  # 按相同日历季度精确连接基线
quarterly['trend_index']=range(len(quarterly))  # 构造趋势特征
model_frame=quarterly.dropna(subset=['baseline']).copy()  # 固定双模型共同样本
if len(model_frame)<12: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 要求至少八季训练与四季验证
validation_size=max(4,len(model_frame)//3)  # 事先设定至少四季度末段验证窗
if len(model_frame)-validation_size<8: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 训练窗检查要求
training,validation=model_frame.iloc[:-validation_size],model_frame.iloc[-validation_size:].copy()  # 按时间切分
model=LinearRegression().fit(training[['trend_index']],training['revenue'])  # 仅在训练窗拟合
validation['trend_forecast']=model.predict(validation[['trend_index']])  # 生成样本外预测
validation['baseline_ape']=(validation['revenue']-validation['baseline']).abs()/validation['revenue'].abs()  # 计算基线误差
validation['trend_ape']=(validation['revenue']-validation['trend_forecast']).abs()/validation['revenue'].abs()  # 计算趋势误差
if not np.isfinite(validation[['baseline_ape','trend_ape']].to_numpy()).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 非有限误差停止
decision={'threshold':0.15,'threshold_source':'课堂误差情景；实际应用应结合企业预算确定','owner':'农业企业经营负责人','action':'趋势MAPE优于同一时期的季节基准且不超过设定值时，可小范围试用','recheck_when':'公司组成、字段、收入计算方法改变，或缺少季度、MAPE超过设定值','recheck_date':'下一季度财报披露后第5个工作日'}  # 记录比较结果和建议
decision['baseline_mape']=float(validation['baseline_ape'].mean()); decision['trend_mape']=float(validation['trend_ape'].mean())  # 发布同一验证窗双模型误差
decision['status']='仅作课堂比较' if decision['trend_mape']<decision['baseline_mape'] and decision['trend_mape']<=decision['threshold'] else '暂不采用'  # 仅允许小范围试用或暂停
structure=panel.groupby(['end_date','area']).agg(companies=('ts_code','nunique'),revenue=('quarter_revenue','sum'),median_margin=('operating_margin','median')).reset_index()  # 汇总固定样本的地区结构
print(version_audit,cohort_audit,validation,decision,structure)  # 输出版本、统计口径、样本组成、模型比较与产品结构结果